package xin.marcher.module.inventory.manage;

import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import xin.marcher.module.common.constants.inventory.TairInventoryConstant;
import xin.marcher.module.common.enums.inventory.BucketOperateEnum;
import xin.marcher.module.common.enums.inventory.BucketOperateStatusEnum;
import xin.marcher.module.common.tair.TairCache;
import xin.marcher.module.inventory.cache.InventoryBucketCache;
import xin.marcher.module.inventory.domain.bo.InventoryBucketOperateBO;
import xin.marcher.module.inventory.strategy.route.BucketOperateStrategyFactory;

import java.util.concurrent.BlockingQueue;

/**
 * 多线程消费队列内的数据
 */
@Slf4j
public class OperateRunner implements Runnable {

    /**
     * 处理任务的队列
     */
    private BlockingQueue blockingQueue;

    private TairCache tairCache;

    private InventoryBucketCache inventoryBucketCache;

    private BucketOperateStrategyFactory operateFactory;

    public OperateRunner(BlockingQueue blockingQueue, TairCache tairCache, InventoryBucketCache inventoryBucketCache, BucketOperateStrategyFactory operateFactory) {
        this.blockingQueue = blockingQueue;
        this.tairCache = tairCache;
        this.inventoryBucketCache = inventoryBucketCache;
        this.operateFactory = operateFactory;
    }

    /**
     * 内部线程处理每个队列的数据
     */
    @Override
    public void run() {
        while (true) {
            if (CollectionUtils.isEmpty(blockingQueue)) {
                try {
                    Thread.sleep(500);
                    continue;
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            InventoryBucketOperateBO operate = null;
            try {
                operate = (InventoryBucketOperateBO) blockingQueue.take();
            } catch (InterruptedException e) {
                log.error("获取分桶任务异常", e);
            }

            if (operate == null) {
                continue;
            }

            try {
                // 尝试将分桶操作设置成处理中，成功了则继续处理，失败了表示已经处理了，则直接跳过
                if (!tairCache.setNx(TairInventoryConstant.BUCKET_OPERATE_PREFIX + operate.getId(),
                        BucketOperateStatusEnum.PROCESS.getCodeString(), 0)) {
                    // 任务不需要执行，减少任务数量
                    inventoryBucketCache.decrOperateCount(TairInventoryConstant.BUCKET_OPERATE_PROCESS_COUNT);
                    continue;
                }

                BucketOperateEnum operateEnum = BucketOperateEnum.getByCode(operate.getOperateType());
                if (operateEnum != null) {
                    operateFactory.getStrategy(operateEnum.getService()).process(operate);
                }

                // 处理完成的将状态改为成功
                tairCache.set(TairInventoryConstant.BUCKET_OPERATE_PREFIX + operate.getId(),
                        BucketOperateStatusEnum.FINISH.getCodeString(), 0);

                // 把成功的id放到缓存中
                inventoryBucketCache.addOperateId(TairInventoryConstant.BUCKET_OPERATE_PROCESS, operate.getId());
                // 任务执行后，减少任务数量
                inventoryBucketCache.decrOperateCount(TairInventoryConstant.BUCKET_OPERATE_PROCESS_COUNT);
            } catch (Exception e) {
                // 处理分桶任务异常，将任务处理状态删除，不然后续再处理，setNx去设置状态为处理中会失败。
                tairCache.delete(TairInventoryConstant.BUCKET_OPERATE_PREFIX + operate.getId());
                // 任务执行失败的，也减少任务数量
                inventoryBucketCache.decrOperateCount(TairInventoryConstant.BUCKET_OPERATE_PROCESS_COUNT);
                log.error("处理分桶任务异常", e);
            }
        }
    }


}
